package com.binance.notification.web.kafka;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;

import java.util.Collections;

@EnableBinding(IChannel.class)
@Slf4j
public class Productor {

    @Autowired
    private BinderAwareChannelResolver resolver;

    @Autowired
    private IChannel channel;

    public void sendMessage(String channel, Object message) {
        Message<String> msg = MessageBuilder.createMessage(JSON.toJSONString(message),
                new MessageHeaders(Collections.singletonMap(MessageHeaders.CONTENT_TYPE, "application/json")));
        log.info("发送信息到Kafka：{}", msg);
        resolver.resolveDestination(channel).send(msg);
    }

    /**
     * 价格提醒，发送交易对信息到kafka
     *
     * @param message
     */
    public void sendAlertPricePairMessage(Object message) {
        Message<String> msg = MessageBuilder.createMessage(JSON.toJSONString(message),
                new MessageHeaders(Collections.singletonMap(MessageHeaders.CONTENT_TYPE, "application/json")));
        log.info("价格提醒：发送交易对和最新价格到Kafka：{}", msg);
        channel.alertPricePairOutput().send(msg);
    }

    /**
     * 价格提醒，发送待推送价格提醒信息到kafka
     *
     * @param message
     */
    public void sendAlertPriceMessage(Object message) {
        Message<String> msg = MessageBuilder.createMessage(JSON.toJSONString(message),
                new MessageHeaders(Collections.singletonMap(MessageHeaders.CONTENT_TYPE, "application/json")));
        log.info("价格提醒：发送待推送信息到Kafka：{}", msg);
        channel.alerPriceOutput().send(msg);
    }

}
